use crate::data_frame::value::{ElementValue, ConfValue, ElementRealConfValue};
use crate::conf_init::{MapResult, MapErr};
use sapeur::{Parser, parse_text, ParserResult};
use sapeur::text::{TextChar, text_key_world, text_just, text_ident};
use sapeur::recursive::{Recursive, recursive};
use crate::logical_plan::expr::func::{LogicalPlanExprFuncOrValue, logical_plan_expr_func_map_to_run_plan_expr_func};
use crate::run_plan::window::RunPlanExprPartitionType;

#[derive(Debug,Clone)]
pub enum LogicalPlanExprPartitionType {
    Order(Vec<LogicalPlanExprFuncOrValue>),
    DisOrder(Vec<LogicalPlanExprFuncOrValue>),
}

fn parse_logical_expr_window_partition_by<'a>()
    ->impl Parser<char, Vec<LogicalPlanExprFuncOrValue>, TextChar>+'a{

    let recusive_parser: Recursive<'a,char, LogicalPlanExprFuncOrValue,TextChar> =

        recursive::<'a,char, LogicalPlanExprFuncOrValue,TextChar,_,_>(|parser| {

            let parse_just_conf = text_key_world()
                .map_end(|o|
                    LogicalPlanExprFuncOrValue::Value(ElementRealConfValue::ConfValue(ConfValue::new(o)))
                );

            let parse_just_element = text_key_world()
                .separated_by(text_just(","), false)
                .delimited_by(text_just("("), text_just(")"))
                .map_end(|o|
                    LogicalPlanExprFuncOrValue::Value(ElementRealConfValue::ElementValue(ElementValue::new(o)))
                );

            let parse_func_element = text_ident().and(
                parser.clone()
                    .separated_by(text_just(","), false)
                    .delimited_by(text_just("("), text_just(")"))
            )
                .map_end(|(n,o)|
                    LogicalPlanExprFuncOrValue::FuncWithValue(n, o)
                );

            parse_just_conf.or(parse_just_element).or(parse_func_element)
        });

    recusive_parser.separated_by(text_just(" "), false)
        .map_end(move |o|  o)
}

pub fn parse_partition_by(order: bool, desc: &str)->MapResult<LogicalPlanExprPartitionType>{
    let res: ParserResult<Vec<LogicalPlanExprFuncOrValue>,_> = parse_text(desc,
                                                                          &parse_logical_expr_window_partition_by(),
                                                                          vec!["no"].as_slice());
    match res {
        ParserResult::OkEnd(t) => {
            return MapResult::Ok(
                if order {
                    LogicalPlanExprPartitionType::Order(t)
                } else {
                    LogicalPlanExprPartitionType::DisOrder(t)
                }
            );
        },
        ParserResult::Ok(_) => {
            return MapResult::Err(MapErr::new(format!("parse_partition_by [{}] parse Ok not to OkEnd", desc)));
        },
        ParserResult::Err(e) => {
            return MapResult::Err(MapErr::new(format!("parse_partition_by [{}] Err at offset [{}]",
                desc, e.offset)))
        },
        ParserResult::ErrEnd(e) => {
            return MapResult::Err(MapErr::new(format!("parse_partition_by [{}] ErrEnd at offset [{}]",
                                                      desc, e.offset)))
        },
    }
}

pub fn partition_by_map(partition_by: &Option<LogicalPlanExprPartitionType>)
    ->MapResult<Option<RunPlanExprPartitionType>>{

    if let Some(partition_by) = partition_by{
        let (order,t) = match partition_by{
            LogicalPlanExprPartitionType::Order(t) => {(true,t)},
            LogicalPlanExprPartitionType::DisOrder(t) => {(false,t)},
        };
        let mut list = Vec::new();
        for t in t.iter(){
            let tmp = logical_plan_expr_func_map_to_run_plan_expr_func(t)?;
            list.push(tmp);
        }

        return MapResult::Ok(Some(
            if order{
                RunPlanExprPartitionType::Order(list)
            }else{
                RunPlanExprPartitionType::DisOrder(list)
            }
        ));
    }else{
        return MapResult::Ok(None);
    }
}

#[test]
fn partition_by_conf_map_test(){
    let r = log4rs_init_local();
    println!("log4rs init {:?}", r);

    let desc = "5 (a,b,c) element_pre((d,e,f),6)";
    let res = parse_text(desc,
                         parse_logical_expr_window_partition_by(),
                         vec!["no"].as_slice());
    println!("res: [{:?}]", res);
}
